package com.learn.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

/**
 * Kafka 消息生产者
 *
 * @Author: cong zhi
 * @CreateDate: 2021/3/2 15:37
 * @UpdateUser: cong zhi
 * @UpdateDate: 2021/3/2 15:37
 * @UpdateRemark: 修改内容
 * @Version: 1.0
 */
public class KafkaProducerDemo extends Thread {

    /**
     * producer api
     */
    private final KafkaProducer<Integer, String> producer;
    /**
     * topic
     */
    private final String topic;
    /**
     * 同步/异步
     */
    private final Boolean isAsync;

    public KafkaProducerDemo(String topic, Boolean isAsync) {
        Properties properties = new Properties();
        // kafka 集群地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.110.17:9092");
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerDemo");
        // kafka中默认都是批量发送
        // 设置批次大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
        // 设置5ms
        properties.put(ProducerConfig.LINGER_MS_CONFIG,5000);
        // 设置压缩
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
        // 设置broke ACK应答机制
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        // 设置key序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 指定自定义分区
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.learn.kafka.MyPartition");
        producer = new KafkaProducer<Integer, String>(properties);
        this.topic = topic;
        this.isAsync = isAsync;
    }

    @Override
    public void run() {
        int num = 0;
        while (true) {
            String message = "kafka:practice:message:" + num;
            System.out.println("begin send message:" + message);
            if (isAsync) { // 异步发送
                //  使用临时变量 finalNum 解决匿名内部类访问外部变量的问题
                final int finalNum = num;
                // 回调
                producer.send(new ProducerRecord<Integer, String>(topic, finalNum, message), (recordMetadata, e) -> {
                    if (null != recordMetadata) {
                        System.out.println("async-offset:" + recordMetadata.offset() +
                                "-> partition:" + recordMetadata.partition() +
                                "-> topic:" + recordMetadata.topic() +
                                "-> key:" + finalNum);
                    }
                });
            } else { // 同步发送
                try {
                    RecordMetadata recordMetadata = producer.send(new ProducerRecord<Integer, String>(topic, num, message)).get();
                    System.out.println("sync-offset:" + recordMetadata.offset() +
                            "-> partition" + recordMetadata.partition() +
                            "-> topic:" + recordMetadata.topic() +
                            "-> key:" + num);
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }

            }
            ++num;
        }
    }

    public static void main(String[] args) {
        new KafkaProducerDemo("test-lisa", false).start();
    }
}
